#include "thread_pool.h"
#include "utils.h"
#include "logger.h"


ThreadPool_t *createThreadPool(int thread_count)
{
    ThreadPool_t *pool = (ThreadPool_t *)calloc(1, sizeof(ThreadPool_t));
    if (pool == NULL)
    {
        PRINT_ERROR("calloc pool");
        return NULL;
    }

    pool->thread_ids = (pthread_t *)calloc(thread_count, sizeof(pthread_t));
    if (pool->thread_ids == NULL)
    {
        PRINT_ERROR("calloc thread_ids");
        free(pool);
        return NULL;
    }

    pool->task_queue = createTaskQueue();

    // 初始化锁和条件变量必须在创建子线程前
    pthread_mutex_init(&pool->lock, NULL);
    pthread_cond_init(&pool->cond, NULL);

    for (int i = 0; i < thread_count; i++)
    {
        if (pthread_create(&pool->thread_ids[i], NULL, threadWorker, (void *)pool) != 0)
        {
            PRINT_ERROR("pthread_create failed");
            free(pool->thread_ids);
            free(pool);
            return NULL;
        }
        pool->active_threads++;
    }

    pool->threads_num = thread_count;

    LOG_INFO("ThreadPool", "ThreadPool created! %d threads active", pool->active_threads);

    return pool;
}

void destroyThreadPool(ThreadPool_t *pool)
{
    pthread_mutex_lock(&(pool->lock));
    pool->shutdown = 1; // 修改线程池关闭标记位
    pthread_cond_broadcast(&(pool->cond));
    pthread_mutex_unlock(&(pool->lock));

    // 等待所有工作线程退出
    for (int i = 0; i < pool->threads_num; ++i)
    {
        LOG_DEBUG("ThreadPool", "Waiting for thread-%ld to exit...", pool->thread_ids[i]);
        //PRINT_DEBUG("Waiting for thread-%ld to exit...", pool->thread_ids[i]);
        pthread_join(pool->thread_ids[i], NULL);
    }

    if (pool->task_queue->size != 0)
    {
        PRINT_DEBUG("%d tasks unfinished!", pool->task_queue->size);
        LOG_WARNING("ThreadPool", "%d tasks unfinished!", pool->task_queue->size);
    }
    destroyTaskQueue(pool->task_queue);

    pthread_mutex_destroy(&(pool->lock));
    pthread_cond_destroy(&(pool->cond));
    free(pool->thread_ids);
    free(pool);

    PRINT_INFO("ThreadPool destroyed!");
    LOG_INFO("ThreadPool", "ThreadPool destroyed!");
}

void submitTask(ThreadPool_t *pool, void (*function)(void *), void *arg)
{
    Task_t *task = (Task_t *)malloc(sizeof(Task_t)); // 这里申请内存，在子线程执行完后才释放
    task->function = function;
    task->arg = arg;
    task->next = NULL;

    pthread_mutex_lock(&(pool->lock));
    enqueue(pool->task_queue, task);
    pthread_cond_signal(&(pool->cond));
    pthread_mutex_unlock(&(pool->lock));
}

void *threadWorker(void *arg)
{
    ThreadPool_t *pool = (ThreadPool_t *)arg;

    while (1)
    {
        pthread_mutex_lock(&pool->lock);

        // 等待直到任务队列非空或线程池关闭
        while (pool->task_queue->size == 0 && !pool->shutdown)
        {
            pthread_cond_wait(&pool->cond, &pool->lock);
        }

        // 如果线程池已经关闭，则子线程退出
        if (pool->shutdown)
        {
            LOG_DEBUG("ThreadPool", "son thread-%lu exit", pthread_self());
            // PRINT_DEBUG("son thread-%lu exit", pthread_self());
            pthread_mutex_unlock(&pool->lock);
            pthread_exit(NULL);
        }

        Task_t *task = dequeue(pool->task_queue);
        pthread_mutex_unlock(&pool->lock);

        (*(task->function))(task->arg);
        PRINT_DEBUG("task is finished!");
        // free(task->arg); // 执行结束释放传进来的参数
        free(task); // 任务函数执行结束，这里释放任务结构体
    }
    return NULL;
}